package com.example.mq.config.mq;

import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.UnsupportedEncodingException;
import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

@Component
public class MqProducerService {

    private static final Logger logger = LoggerFactory.getLogger(MqProducerService.class);

    private final Lock lock = new ReentrantLock();
    @Autowired
    private DefaultMQProducer producer;

    public MqProducerService() throws MQClientException {

    }

    public void sendAsync(String topic, String tag, String msg) throws UnsupportedEncodingException, RemotingException, InterruptedException, MQClientException {
        sendMessage(new Message(topic, tag, msg.getBytes(StandardCharsets.UTF_8)));
    }

    public void sendMessage(Message message) throws UnsupportedEncodingException, MQClientException, RemotingException, InterruptedException {
        producer.send(message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                logger.info("异步调用成功",sendResult);
            }

            @Override
            public void onException(Throwable throwable) {
                logger.error(throwable.getLocalizedMessage());
            }
        });
    }

    public SendResult sendSync(String topic, String tag, String msg) throws UnsupportedEncodingException, MQBrokerException, RemotingException, InterruptedException, MQClientException {
        return getSendResult(new Message(topic, tag, msg.getBytes(RemotingHelper.DEFAULT_CHARSET)));
    }

    public SendResult getSendResult(Message message) throws UnsupportedEncodingException, MQClientException, RemotingException, MQBrokerException, InterruptedException {
        SendResult send = producer.send(message);
        return send;
    }

    public SendResult sendDelay(Integer level, Message message){
        message.setDelayTimeLevel(level);
        SendResult sendResult = null;
        try {
            sendResult = producer.send(message);
        } catch (Exception e) {
            e.printStackTrace();
        }
        return sendResult;
    }

    public SendResult batchSend(List<Message> messageList){
        SendResult sendResult = null;
        try {
            sendResult = producer.send(messageList);
        }catch (Exception e){
            e.printStackTrace();
        }
        return sendResult;
    }

    public void batchSendAsync(List<Message> messageList){
        try {
            producer.send(messageList, new SendCallback() {
                @Override
                public void onSuccess(SendResult sendResult) {
                    logger.info("消费端已经成功消费消息");
                }

                @Override
                public void onException(Throwable throwable) {
                    logger.error("消费过程中出现异常:"+throwable.getLocalizedMessage());
                }
            });
        }catch (Exception e){
            e.printStackTrace();
        }
    }
}
